package com.hxk.merge

import com.hxk.spark.merge.LogMessage
import org.apache.commons.lang3.time.FastDateFormat

import scala.collection.mutable.ArrayBuffer


/* Take User Message as an example
{"user_id": "UserA", "location": "LocationA", "start_time": "2018-01-01 08:00:00", "residence_time": 60},
{"user_id": "UserA", "location": "LocationA", "start_time": "2018-01-01 09:00:00", "residence_time": 60},
{"user_id": "UserA", "location": "LocationB", "start_time": "2018-01-01 10:00:00", "residence_time": 60},
{"user_id": "UserA", "location": "LocationA", "start_time": "2018-01-01 11:00:00", "residence_time": 60}
*/

class OneUserGroupProcess(logMessagess: Array[LogMessage]) {

  //根据start_time时间进行排序（升序，一次排序即可，所以放到成员变量中）
  private val sortedLogMessage = logMessagess.sortBy(logMessagess => logMessagess.getStartTime.toString)

  //LogMessage中的时间都是String类型，不好比较时间，需要转成Long类型 TimeStep
  private val dataFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")

  /**
    * 对当前User的所有Message进行time merge
    * @return
    */
  def timeMerge(): ArrayBuffer[LogMessage] = {

    //用于存放当前组的LogMessage
    val oneUserGroupMessage = new ArrayBuffer[LogMessage]()

    //实现Time Merge 底层操作
    //如果location相同，并且current条start_time减去前一条start_time等于前一条的residence_time
    //那么就把前一条residence_time加到current条residence_time上
    sortedLogMessage.foldLeft(Option.empty[LogMessage]) { case (preLog, currLog) =>
      val currLofStartTime = dataFormat.parse(currLog.getStartTime.toString).getTime
      if(preLog.nonEmpty && currLog.getLocation.toString == preLog.get.getLocation.toString &&
        currLofStartTime - dataFormat.parse(preLog.get.getStartTime.toString).getTime ==
          (preLog.get.getResidenceTime)*60*1000)  {

        //Time Merge : 把前一条residence_time加到current条residence_time上
        currLog.setResidenceTime(currLog.getResidenceTime + preLog.get.getResidenceTime)
        currLog.setStartTime(preLog.get.getStartTime)
        oneUserGroupMessage -= preLog.get

      }
      oneUserGroupMessage += currLog

      Some(currLog)
    }


    oneUserGroupMessage
  }
}
